package com.tianma.component;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.Message;
import com.tianma.sync.AbstractSyncTemplate;
import com.tianma.sync.TradeInfoSync;
import com.tianma.sync.TradeOrdersPickingSync;
import com.tianma.sync.TradeOrdersSync;
import com.tianma.util.CanalDataParser;
import com.tianma.model.InnerBinlogEntry;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

@Component
public class CanalClient {
    @Resource(name = "canalSimpleConnector")
    private CanalConnector canalConnector;

    @Autowired
    private TradeOrdersPickingSync tradeOrdersPickingSync;

    @Autowired
    private TradeOrdersSync tradeOrdersSync;

    @Autowired
    private TradeInfoSync tradeInfoSync;

    public void start() {
        int batchSize = 5 * 1024;
        try {
            canalConnector.connect();
//            canalConnector.subscribe(".*\\..*");
            canalConnector.subscribe(null);
            canalConnector.rollback();


            while (true){
                Message message = canalConnector.getWithoutAck(batchSize);

                long messageId = message.getId();
                int size = message.getEntries().size();

                if(messageId == -1 || size == 0){
                    try {
                        Thread.sleep(3000);
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }else{
                    //进行数据同步
                    //1. 解析Message对象
                    List<InnerBinlogEntry> innerBinlogEntries = CanalDataParser.convertToInnerBinlogEntry(message);

                    //2. 将解析后的数据信息 同步到其它MySQL中
                    syncDataToMysql(innerBinlogEntries);
                }

                //提交确认
                canalConnector.ack(messageId);
            }

        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            canalConnector.disconnect();
        }
    }

    private void syncDataToMysql(List<InnerBinlogEntry> innerBinlogEntries) throws Exception {

        //遍历数据集合 , 根据数据集合中的数据信息, 来决定执行增加, 修改 , 删除操作 .
        if(innerBinlogEntries != null){
            for (InnerBinlogEntry innerBinlogEntry : innerBinlogEntries) {
                // 不是tianma_gx_data的数据直接忽略
                if (!StringUtils.equalsIgnoreCase(innerBinlogEntry.getSchemaName(), "tianma_gx_data")) {
                    continue;
                }
                AbstractSyncTemplate syncTemplate;
                if (StringUtils.equalsIgnoreCase(innerBinlogEntry.getTableName(), "trade_orders_picking")) {
                    syncTemplate = tradeOrdersPickingSync;
                } else if (StringUtils.equalsIgnoreCase(innerBinlogEntry.getTableName(), "trade_orders")) {
                    syncTemplate = tradeOrdersSync;
                } else if (StringUtils.equalsIgnoreCase(innerBinlogEntry.getTableName(), "trade_info")) {
                    syncTemplate = tradeInfoSync;
                } else {
                    continue;
                }
                syncTemplate.sync(innerBinlogEntry);

            }
        }
    }

}
